package com.zzl.mq.storage;

import com.zzl.mq.network.gossip.DataMessagePayload;
import org.springframework.stereotype.Component;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 基于内存的消息存储实现
 */
@Component
public class MemoryStorage implements Storage {
    
    private final Map<String, List<DataMessagePayload>> messageStore = new ConcurrentHashMap<>();
    
    @Override
    public boolean store(String topic, DataMessagePayload message) {
        messageStore.computeIfAbsent(topic, k -> new CopyOnWriteArrayList<>())
                   .add(message);
        return true;
    }
    
    @Override
    public List<DataMessagePayload> fetch(String topic, long offset, int limit) {
        List<DataMessagePayload> messages = messageStore.getOrDefault(topic, new ArrayList<>());
        int fromIndex = (int) Math.min(offset, messages.size());
        int toIndex = Math.min(fromIndex + limit, messages.size());
        return messages.subList(fromIndex, toIndex);
    }
    
    @Override
    public int deleteMessagesBefore(String topic, long beforeTime) {
        List<DataMessagePayload> messages = messageStore.get(topic);
        if (messages == null) {
            return 0;
        }
        
        AtomicInteger count = new AtomicInteger(0);
        messages.removeIf(message -> {
            if (message.getTimestamp() < beforeTime) {
                count.incrementAndGet();
                return true;
            }
            return false;
        });
        return count.get();
    }
    
    @Override
    public long getMessageCount(String topic) {
        return messageStore.getOrDefault(topic, new ArrayList<>()).size();
    }
} 